package time.eventTime;

import api.beans.SensorReading;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.timestamps.BoundedOutOfOrdernessTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * 由DataStream转换成表时指定（推荐）
 */
public class eventTime1 {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);

        String path = "D:\\IdeaProjects\\springboot-flink-1\\flinkTutorial\\src\\main\\resources\\sensor.txt";
        DataStream<String> inputStream = env.readTextFile(path);
        DataStream<SensorReading> dataStream = inputStream.map(line -> {
            String[] fields = line.split(",");
            return new SensorReading(fields[0],new Long(fields[1]), new Double(fields[2]));
        }).assignTimestampsAndWatermarks(new BoundedOutOfOrdernessTimestampExtractor<SensorReading>(Time.seconds(2)){
            @Override
            public long extractTimestamp(SensorReading el) {
                return el.getTimestamp() * 1000L;
            }
        });
        Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp.rowtime as ts , temperature as temp");
        // 直接追加时间字段
        //Table sensorTable = tableEnv.fromDataStream(dataStream, "id, timestamp as ts , temperature as temp, rt.rowtime");
        // 输出结构
        sensorTable.printSchema();
        // 打印输出
        tableEnv.toAppendStream(sensorTable, Row.class).print();

        env.execute();
    }
}
